package com.zhxy.gps.gateway;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;

import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.zhxy.gps.terminal.Terminal;
import com.zhxy.gps.terminal.message.AbstractMessage;
import com.zhxy.gps.terminal.message.AccessMessage;
import com.zhxy.gps.terminal.message.KeepAliveMessage;
import com.zhxy.gps.terminal.message.LocateMessage;
import com.zhxy.gps.terminal.message.LocationMessage;
import com.zhxy.gps.terminal.message.ReplyMessage;
import com.zhxy.util.async.ResultHandler;
import com.zhxy.util.message.AsyncMessageProcessor;
import com.zhxy.util.message.DuplexMessagePipe;
import com.zhxy.util.message.DuplexMessageServiceListener;
import com.zhxy.util.message.MessageCodec;
import com.zhxy.util.message.MessagePipe;
import com.zhxy.util.message.MessagePipeIoHandlerConfig;
import com.zhxy.util.message.MinaDuplexMessageService;
import com.zhxy.util.message.jsonpack.JsonpackCodec;

public class SmartphoneGateway implements DuplexMessageServiceListener, AsyncMessageProcessor<DuplexMessagePipe, Object>, TerminalGateway {
	
	private static final Logger logger = LoggerFactory.getLogger(SmartphoneGateway.class);
	
	private Map<String, DuplexMessagePipe> onlines = new ConcurrentHashMap<String, DuplexMessagePipe>();
	
	private NioSocketAcceptor acceptor;
	
	private MinaDuplexMessageService duplexMessageService;
	
	private AsyncMessageProcessor<GatewayMessageContext, AbstractMessage> processor;
	
	private MessageCodec messageCodec;
	
	private ScheduledExecutorService scheduledExecutor;

	@Override
	public void setProcessor(
			AsyncMessageProcessor<GatewayMessageContext, AbstractMessage> processor) {
		this.processor = processor;
	}

	public void setMessageCodec(MessageCodec messageCodec) {
		this.messageCodec = messageCodec;
	}

	public void setScheduledExecutor(ScheduledExecutorService scheduledExecutor) {
		this.scheduledExecutor = scheduledExecutor;
	}

	public void bind(InetSocketAddress address) throws IOException {
		
		MessagePipeIoHandlerConfig config = new MessagePipeIoHandlerConfig();
		config.setTimeout(30);
		
		acceptor = new NioSocketAcceptor();
		
		duplexMessageService = new MinaDuplexMessageService();
		duplexMessageService.setConfig(config);
		duplexMessageService.setCodec(messageCodec);
		duplexMessageService.addMessageProcessor(this);
		duplexMessageService.setScheduledExecutor(scheduledExecutor);
		duplexMessageService.addDuplexMessageServiceListener(this);
		duplexMessageService.bindIoService(acceptor);
		
		acceptor.bind(address);
	}
	
	public void close() {
		if (acceptor != null) {
			acceptor.dispose(true);
			acceptor = null;
		}
	}
	
	@Override
	public void send(String deviceSerial, AbstractMessage message, ResultHandler<ReplyMessage> handler) {
		TerminalSession session = getSession(deviceSerial);
		if (session == null) {
			handler.exception(new Exception("终端:" + deviceSerial + "不在线"));
			return;
		}
		
		class Handler implements ResultHandler<Object> {
			private ResultHandler<ReplyMessage> handler;
			public Handler(ResultHandler<ReplyMessage> handler) {
				this.handler = handler;
			}
			public void result(Object result) {
				handler.result((ReplyMessage)result);
			}
			public void exception(Throwable e) {
				handler.exception(e);
			}
		}
		
		session.getMessagePipe().request(message, new Handler(handler));
	}
	

	@Override
	public Object status(String deviceSerial, String status) {
		
		if (STATUS_ONLINE.equals(status)) {
			return onlines.containsKey(deviceSerial);
		}
		
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public void connected(DuplexMessagePipe pipe) {
		TerminalSession session = new TerminalSession();
		session.setMessagePipe(pipe);
		pipe.setAttribute(TerminalSession.class.getName(), session);
	}

	@Override
	public void disconnected(DuplexMessagePipe connection) {
		
		TerminalSession session = getSession(connection);
		if (session == null)
			return;

	}
	
	@Override
	public void process(DuplexMessagePipe pipe, Object message,
			ResultHandler<Object> handler) {
		
		TerminalSession session = getSession(pipe);
		//AbstractMessage msg = (AbstractMessage)message;
		
		// 收到终端的接入消息
		if (message instanceof AccessMessage) {
			AccessMessage access = (AccessMessage)message;
			session.setTerminal(new Terminal(access.getDeviceSerial()));
			onlines.put(access.getDeviceSerial(), pipe);
			handler.result(new ReplyMessage());
			
			logger.info("验证终端完毕:" + access.getDeviceSerial());
		} else if (message instanceof KeepAliveMessage) {
		} else {
			GatewayMessageContext context = new GatewayMessageContext();
			context.setMessagePipe(pipe);
			context.setTerminal(session.getTerminal());
			context.setGateway(this);
			
			class Handler implements ResultHandler<AbstractMessage> {
				private ResultHandler<Object> handler;
				public Handler(ResultHandler<Object> handler) {
					this.handler = handler;
				}
				public void result(AbstractMessage result) {
					handler.result(result);
				}
				public void exception(Throwable e) {
					handler.exception(e);
				}
			}
			
			processor.process(context, (AbstractMessage)message, new Handler(handler));
		}
	}
	
	protected synchronized TerminalSession getSession(String deviceSerial) {
		MessagePipe pipe = onlines.get(deviceSerial);
		if (pipe == null)
			return null;
		return getSession(pipe);
	}
	
	protected synchronized TerminalSession getSession(MessagePipe pipe) {
		return (TerminalSession)pipe.getAttribute(TerminalSession.class.getName());
	}
	
	/**
	 * 如果存在，关闭会话
	 * @param sessionId
	 */
	protected void closeSession(MessagePipe sessionId) {
		
		TerminalSession session = getSession(sessionId);
		if (session == null)
			return;

	}

	
	public class TerminalSession {
		
		private Terminal terminal;
		
		private DuplexMessagePipe messagePipe;

		public TerminalSession() {
			
		}

		public Terminal getTerminal() {
			return terminal;
		}

		public void setTerminal(Terminal terminal) {
			this.terminal = terminal;
		}

		public DuplexMessagePipe getMessagePipe() {
			return messagePipe;
		}

		public void setMessagePipe(DuplexMessagePipe messagePipe) {
			this.messagePipe = messagePipe;
		}
	}
	
	
	public static MessageCodec createCodec() {
		JsonpackCodec codec = new JsonpackCodec();
		codec.registerType("locate", LocateMessage.class);
		codec.registerType("access", AccessMessage.class);
		codec.registerType("location", LocationMessage.class);
		codec.registerType("reply", ReplyMessage.class);
		
		return codec;
	}

}
